package felix.things.router.mqtt.sender;

import felix.things.router.mqtt.MqttBrokerMonitor;
import felix.things.router.mqtt.persistent.MqttPersistentMessage;
import felix.things.router.mqtt.persistent.PersistenceConfiguration;
import felix.things.router.mqtt.persistent.PersistentFileService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

/**
 * @author Felix
 * @date 2019/7/29
 */
@Slf4j
@Component
public class MqttMessageSender implements Runnable {

    @Autowired
    private MqttBrokerMonitor mqttBroker;
    @Autowired
    private PersistentFileService persistentFileService;
    @Autowired
    private PersistenceConfiguration persistence;

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                mqttBroker.checkConnection();
                List<MqttPersistentMessage> storedMessages = getMessages();
                if (!storedMessages.isEmpty()) {
                    for (MqttPersistentMessage message : storedMessages) {
                        log.debug("Sending message 【{}】", message);
                        mqttBroker.publish(message);
                    }
                } else {
                    Thread.sleep(persistence.getPollingInterval());
                }
            } catch (InterruptedException e) {
                log.trace(e.getMessage());
                Thread.currentThread().interrupt();
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private List<MqttPersistentMessage> getMessages() {
        try {
            List<MqttPersistentMessage> resendMessages = persistentFileService.getResendMessages();
            if (!resendMessages.isEmpty()) {
                return resendMessages;
            } else {
                return persistentFileService.getPersistentMessages();
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }
}
